package cn.com.sparkle.paxos.net.callback;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.LookUpLatestInstanceIdResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.net.CallBack;
import cn.com.sparkle.paxos.net.NetNode;

//import cn.com.sparkle.paxos.net.message.LookUpLatestInstanceIdResponse;

public abstract class LookUpLatestInstanceIdCallBack implements CallBack {
	private final static Logger logger = Logger
			.getLogger(LookUpLatestInstanceIdCallBack.class);

	private int needResponseCount;
	private final int quorum;

	private int goodNum = 0;
	private long maxInstanceId = -2;

	private ReentrantLock lock = new ReentrantLock();

	public LookUpLatestInstanceIdCallBack(int quorum, int needResponseCount) {
		this.quorum = quorum;
		this.needResponseCount = needResponseCount;
	}

	@Override
	public void call(NetNode nnode, MessagePackage value) {
		try {
			lock.lock();
			LookUpLatestInstanceIdResponse response = value
					.getLookUpLatestInstanceIdResponse();
			if (maxInstanceId < response.getInstanceId()) {
				maxInstanceId = response.getInstanceId();
			}
			++goodNum;
			--needResponseCount;
			checkCount();

		} finally {
			lock.unlock();
		}
	}

	public abstract void finish(long instanceId);

	public long getMaxInstanceId() {
		try {
			lock.lock();
			return maxInstanceId;
		} finally {
			lock.unlock();
		}

	}

	private void checkCount() {
		logger.debug("goodNum:" + goodNum + " needResponseCount:"
				+ needResponseCount);
		if (goodNum == quorum) {
			++goodNum;// avoid invoking repeatedly
			finish(maxInstanceId);
		} else if (goodNum < quorum && needResponseCount == 0) {
			maxInstanceId = -2;// assign maxInstance -2 to manifest the number
								// of response is not enough of quorum
			finish(maxInstanceId);
		}
	}

	@Override
	public void fail(NetNode nnode) {
		try {
			lock.lock();
			--needResponseCount;
			checkCount();
		} finally {
			lock.unlock();
		}
	}

}
